package com.wish233.register.impl;

import com.wish233.utils.ServiceMetaUtils;
import io.lettuce.core.RedisClient;
import io.lettuce.core.pubsub.RedisPubSubListener;
import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
import io.lettuce.core.pubsub.api.sync.RedisPubSubCommands;

import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author WHH
 * Redis对key缓存的监听
 */
public class RedisWatcher {
    private final Set<String> watchingKeySet = ConcurrentHashMap.newKeySet();
    private final RedisClient redisClient;
    private final RegistryServiceCache cache = RegistryServiceCache.getCache();
    private final StatefulRedisPubSubConnection<String, String> pubSubConnection;

    public RedisWatcher(RedisClient redisClient) {
        this.redisClient = redisClient;
        this.pubSubConnection = redisClient.connectPubSub();
        this.pubSubConnection.addListener(new RedisKeyEventListener());
    }

    /**
     * 监听服务节点key变化
     *
     * @param serviceNodeKey 服务节点key，格式为 "serviceName:version:host:port"
     */
    public void watch(String serviceNodeKey) {
        // 确保只监听一次
        if (watchingKeySet.add(serviceNodeKey)) {
            // 启用Keyspace通知（如果尚未启用）
            enableKeyspaceNotifications();
            // 订阅该key的事件
            RedisPubSubCommands<String, String> sync = pubSubConnection.sync();
            sync.psubscribe("__keyspace@0__:" + serviceNodeKey);
        }
    }

    /**
     * 确保Redis已配置Keyspace通知
     */
    private void enableKeyspaceNotifications() {
        RedisPubSubCommands<String, String> sync = pubSubConnection.sync();
        String config = sync.configGet("notify-keyspace-events").get("notify-keyspace-events");
        if (!config.contains("K") || !config.contains("E")) {
            sync.configSet("notify-keyspace-events", "KE");
        }
    }

    /**
     * 取消监听
     *
     * @param serviceNodeKey 服务节点key
     */
    public void unwatch(String serviceNodeKey) {
        if (watchingKeySet.remove(serviceNodeKey)) {
            pubSubConnection.sync().punsubscribe("__keyspace@0__:" + serviceNodeKey);
        }
    }

    /**
     * 关闭监听
     */
    public void close() {
        pubSubConnection.close();
    }

    /**
     * Redis key事件监听器
     */
    private class RedisKeyEventListener implements RedisPubSubListener<String, String> {
        @Override
        public void message(String channel, String message) {
            // 格式: __keyspace@0__:serviceNodeKey
            String serviceNodeKey = channel.substring("__keyspace@0__:".length());
            // 只处理删除事件
            if ("del".equals(message)) {
                String serviceKey = ServiceMetaUtils.extractServiceKeyFromNodeKey(serviceNodeKey, '/');
                cache.clearCache(serviceKey);
            }
        }

        @Override
        public void message(String pattern, String channel, String message) {
            // 处理模式匹配的消息
            message(channel, message);
        }

        @Override
        public void subscribed(String channel, long count) {
        }

        @Override
        public void psubscribed(String pattern, long count) {
        }

        @Override
        public void unsubscribed(String channel, long count) {
        }

        @Override
        public void punsubscribed(String pattern, long count) {
        }
    }
}